use std::{str::SplitAsciiWhitespace, sync::Arc, time::Instant, vec};

use crate::{
    resp::{Command, Value},
    RedisError, RedisResult,
};
use parking_lot::RwLock as PRWMutex;
use serde::Deserialize;
use smallvec::SmallVec;

use super::{node::NodeSocket, ConnectionState};

pub(crate) struct ClusterConnection {
    pub(crate) state: Arc<ConnectionState>,
    sockets: Vec<Arc<NodeSocket>>,
    pub(crate) shards: Arc<PRWMutex<Vec<ClusterShard>>>,
}

impl ClusterConnection {
    pub(crate) async fn new(
        hosts: Vec<(String, u32)>,
        state: ConnectionState,
    ) -> RedisResult<Self> {
        let state = Arc::new(state);
        let (socket, shards) = Self::try_connect(&state, hosts).await?;
        // 连接主 master 节点
        let mut sockets = vec![];
        for shard in shards.iter() {
            if shard.master.host == socket.host && shard.master.port == socket.port {
                sockets.push(socket.clone());
            } else {
                let socket = Arc::new(NodeSocket::new(
                    shard.master.host.to_string(),
                    shard.master.port,
                ));
                socket.set_connecting();
                let addr = format!("{}:{}", socket.host, socket.port);
                log::debug!("try connect to {addr}, epoch = {}", socket.epcoh_get());
                super::internal_connect(&state, &socket, &addr).await?;
                sockets.push(socket.clone());
            }
        }
        Ok(Self {
            state,
            sockets,
            shards: Arc::new(PRWMutex::new(shards)),
        })
    }

    async fn try_connect(
        state: &Arc<ConnectionState>,
        hosts: Vec<(String, u32)>,
    ) -> RedisResult<(Arc<NodeSocket>, Vec<ClusterShard>)> {
        for (host, port) in hosts {
            match Self::connect(&host, &port, &state).await {
                Ok(result) => {
                    return Ok(result);
                }
                Err(err) => {
                    log::error!("connect failed, try next host, {err}");
                }
            }
        }
        Err(RedisError::Connect(
            "all of the hosts in cluster are invalid".to_string(),
        ))
    }

    async fn connect(
        host: &str,
        port: &u32,
        state: &Arc<ConnectionState>,
    ) -> RedisResult<(Arc<NodeSocket>, Vec<ClusterShard>)> {
        let socket = Arc::new(NodeSocket::new(host.to_string(), *port));
        socket.set_connecting();
        let addr = format!("{}:{}", socket.host, socket.port);
        log::debug!("try connect to {addr}, epoch = {}", socket.epcoh_get());
        super::internal_connect(state, &socket, &addr).await?;
        // get slots info by `CLISTER SLOTS`
        // let cmd = Command::new_without_key("CLUSTER").arg("SLOTS");
        // let rx = socket.send(&cmd.encode()).await?;
        // let val = super::wait_receive_after_send(&state, rx).await?;
        // let cluster_slots = ClusterSlot::read_from_value(val)?;
        // log::debug!("{cluster_slots:?}");
        let cmd = Command::new("CLUSTER").arg("NODES");
        let rx = socket.send(&cmd.encode()).await?;
        let val = super::wait_receive_after_send(&state, rx).await?;
        let shards = ClusterShard::read_from_value(val, host)?;
        Ok((socket, shards))
    }

    pub async fn send_command<'a, T: Deserialize<'a>>(
        &self,
        command: Command<'a>,
    ) -> RedisResult<T> {
        let mut socket = self
            .get_socket_by_command(&command)?
            .ok_or(RedisError::Send(
                "couldn't find any socket by shard".to_string(),
            ))?;
        let bytes = command.encode();
        loop {
            let rx = match socket.send(&bytes).await {
                Ok(rx) => rx,
                Err(err) => {
                    // retry
                    log::error!("{err:?}");
                    self.send_command_retry(socket, &bytes).await?
                }
            };
            let val = super::wait_receive_after_send(&self.state, rx).await?;
            match val {
                Value::BulkError(buf) | Value::SimpleError(buf) => {
                    let str =
                        String::from_utf8(buf).map_err(|err: std::string::FromUtf8Error| {
                            RedisError::new_response("error on read error string", err)
                        })?;
                    if let Some((action, slot, mut host, port)) = read_moved_or_ask(&str) {
                        if host == "127.0.0.1" || host == "" {
                            host = socket.host.to_string();
                        }
                        socket = self.get_socket_by_host(&host, port)?;
                        if action == "MOVED" {
                            // self
                            self.migrate_slot(slot, socket);
                        }
                    } else {
                        return Err(RedisError::Response(str));
                    }
                }
                _ => {
                    return T::deserialize(&val);
                }
            }
        }
    }

    async fn send_command_retry(
        &self,
        socket: &NodeSocket,
        content: &[u8],
    ) -> RedisResult<tokio::sync::oneshot::Receiver<Value>> {
        for i in 0..self.state.retry_on_send_error {
            tokio::time::sleep(super::RETRY_INTERVAL).await;
            if i == self.state.retry_on_send_error - 1 {
                return socket.send(&content).await;
            } else {
                match socket.send(&content).await {
                    Ok(val) => {
                        return Ok(val);
                    }

                    Err(err) => {
                        log::trace!("error on retry {}, {err}", i + 1);
                    }
                }
            }
        }
        Err(RedisError::Other("error on send_command_retry".to_string()))
    }

    pub(crate) async fn send_command_no_reply<'a>(
        &self,
        command: Command<'a>,
        socket_addr: Option<String>,
    ) -> RedisResult<String> {
        let socket = if let Some(socket_addr) = socket_addr {
            self.get_socket_addr(&socket_addr)?
        } else {
            self.get_socket_by_command(&command)?
                .ok_or(RedisError::Send(
                    "couldn't find any socket by shard".to_string(),
                ))?
        };
        let bytes = command.encode();
        match socket.send_no_reply(&bytes).await {
            Ok(_) => {}
            Err(err) => {
                log::error!("{err:?}");
                self.send_command_no_reply_retry(socket, &bytes).await?;
            }
        };
        Ok(format!("{}:{}", socket.host, socket.port))
    }

    async fn send_command_no_reply_retry(
        &self,
        socket: &NodeSocket,
        content: &[u8],
    ) -> RedisResult<()> {
        for i in 0..self.state.retry_on_send_error {
            tokio::time::sleep(super::RETRY_INTERVAL).await;
            if i == self.state.retry_on_send_error - 1 {
                return socket.send_no_reply(&content).await;
            } else {
                match socket.send_no_reply(&content).await {
                    Ok(_) => {
                        return Ok(());
                    }
                    Err(err) => {
                        log::trace!("error on retry no reply {}, {err}", i + 1);
                    }
                }
            }
        }
        Err(RedisError::Other(
            "error on send_command_no_reply_retry".to_string(),
        ))
    }

    pub(crate) fn get_socket_by_command(
        &self,
        command: &Command,
    ) -> RedisResult<Option<&NodeSocket>> {
        match command.name.to_lowercase().as_str() {
            "subscribe" | "unsubscribe" | "info" | "cluster" | "acl" => {
                return Ok(Some(self.get_socket_random()));
            }
            "bgsave" | "mset" | "mget" => unimplemented!(),
            _ => {}
        }
        let mut shard_to_use = None;
        let read_guard = self.shards.read();
        if let Some(key) = command.key {
            let key_slot = hash_slot(key);
            let shard = get_shard_by_slot(&read_guard, key_slot).ok_or(RedisError::Send(
                format!("couldn't find any shard by slot[{key_slot}]"),
            ))?;
            shard_to_use.replace(shard);
        }
        if let Some(keys) = command.keys {
            for m_key in keys {
                let m_slot = hash_slot(m_key);
                let m_shard = get_shard_by_slot(&read_guard, m_slot).ok_or(RedisError::Send(
                    format!("couldn't find any shard by slot[{m_slot}]"),
                ))?;
                if let Some(o_shard) = shard_to_use {
                    if o_shard != m_shard {
                        return Err(RedisError::Send(
                            "the keys are not in the same shard".to_string(),
                        ));
                    }
                } else {
                    shard_to_use.replace(m_shard);
                }
            }
        }
        if let Some(shard) = shard_to_use {
            for socket in self.sockets.iter() {
                if socket.host == shard.master.host && socket.port == shard.master.port {
                    // if socket.host != shard.master.host || socket.port != shard.master.port {
                    return Ok(Some(socket));
                }
            }
            Ok(None)
        } else {
            Ok(Some(self.get_socket_random()))
        }
    }

    pub(crate) fn get_socket_random(&self) -> &NodeSocket {
        let b = Instant::now().elapsed().as_secs() % (self.sockets.len() as u64);
        self.sockets.iter().nth(b as usize).unwrap()
    }

    fn get_socket_addr(&self, addr: &str) -> RedisResult<&NodeSocket> {
        for socket in self.sockets.iter() {
            if format!("{}:{}", socket.host, socket.port) == addr {
                return Ok(&socket);
            }
        }
        return Err(RedisError::Send(
            "couldn't find any socket by addr".to_string(),
        ));
    }

    fn get_socket_by_host(&self, host: &str, port: u32) -> RedisResult<&NodeSocket> {
        for socket in self.sockets.iter() {
            if socket.host == host && socket.port == port {
                return Ok(&socket);
            }
        }
        return Err(RedisError::Send(
            "couldn't find any socket by addr".to_string(),
        ));
    }

    fn migrate_slot(&self, slot_to_remove: u16, to_socket: &NodeSocket) {
        let mut write_guard = self.shards.write();
        for item in write_guard.iter_mut() {
            for (index, val) in item.slots.iter().enumerate() {
                if val == &slot_to_remove {
                    item.slots.remove(index);
                    break;
                }
            }
        }
        for item in write_guard.iter_mut() {
            if item.master.host == to_socket.host && item.master.port == to_socket.port {
                item.slots.push(slot_to_remove);
                break;
            }
        }
    }
}

// #[derive(Debug)]
// struct ClusterSlot {
//     from: usize,
//     to: usize,
//     nodes: SmallVec<[ClusterNodeInfo; 4]>,
// }

// impl ClusterSlot {
//     fn read_from_value(value: Value) -> RedisResult<Vec<Self>> {
//         let mut result = vec![];
//         match value {
//             Value::Array(array) => {
//                 for item in array {
//                     result.push(Self::internal_read_slot(item)?);
//                 }
//             }
//             _ => {
//                 return Err(RedisError::RespDeserialize(
//                     "the type of cluster slot result doesn't be array".to_string(),
//                 ))
//             }
//         }
//         Ok(result)
//     }

//     fn internal_read_slot(value: Value) -> RedisResult<Self> {
//         let array = match value {
//             Value::Array(array) => array,
//             _ => {
//                 return Err(RedisError::RespDeserialize(
//                     "the type of cluster slot result doesn't be array".to_string(),
//                 ))
//             }
//         };
//         let mut array_iter = array.into_iter();
//         let from: usize = match array_iter.next().ok_or(RedisError::RespDeserialize(
//             "couldn't read [slot from] value from cluster slot result".to_string(),
//         ))? {
//             Value::Integer(val) => Ok(val as usize),
//             _ => Err(RedisError::RespDeserialize(
//                 "couldn't parse [slot from] value into usize".to_string(),
//             )),
//         }?;
//         let to = match array_iter.next().ok_or(RedisError::RespDeserialize(
//             "couldn't read [slot to] value from cluster slot result".to_string(),
//         ))? {
//             Value::Integer(val) => Ok(val as usize),
//             _ => Err(RedisError::RespDeserialize(
//                 "couldn't parse [slot to] value into usize".to_string(),
//             )),
//         }?;
//         let mut array_nodes = smallvec::smallvec![];
//         loop {
//             if let Some(cluster_slot_node) = array_iter.next() {
//                 array_nodes.push(ClusterNodeInfo::read_from_value(cluster_slot_node)?);
//             } else {
//                 break;
//             }
//         }
//         if array_nodes.len() == 0 {
//             return Err(RedisError::RespDeserialize(
//                 "the cluster slot nodes is empty".to_string(),
//             ));
//         }
//         Ok(Self {
//             from,
//             to,
//             nodes: array_nodes,
//         })
//     }
// }

// #[derive(Debug)]
// struct ClusterNodeInfo {
//     ip: String,
//     port: u32,
//     node_id: String,
//     host: Option<String>,
// }

// impl ClusterNodeInfo {
//     fn read_from_value(value: Value) -> RedisResult<Self> {
//         let array = match value {
//             Value::Array(array) => array,
//             _ => {
//                 return Err(RedisError::RespDeserialize(
//                     "the type of cluster slot node doesn't be array".to_string(),
//                 ))
//             }
//         };
//         let mut array_iter = array.into_iter();
//         let ip = array_iter
//             .next()
//             .ok_or(RedisError::RespDeserialize(
//                 "couldn't read [ip] from cluster slot node".to_string(),
//             ))?
//             .into_string()?;
//         let port = array_iter
//             .next()
//             .ok_or(RedisError::RespDeserialize(
//                 "couldn't read [port] from cluster slot node".to_string(),
//             ))?
//             .read_as_u32()?;
//         let node_id = array_iter
//             .next()
//             .ok_or(RedisError::RespDeserialize(
//                 "couldn't read [node_id] from cluster slot node".to_string(),
//             ))?
//             .into_string()?;
//         let host = match array_iter.next().ok_or(RedisError::RespDeserialize(
//             "couldn't read [map] from cluster slot node".to_string(),
//         ))? {
//             Value::Map(mut map) => {
//                 if map.len() > 0 {
//                     let (_, h) = map.remove(0);
//                     Some(h.into_string()?)
//                 } else {
//                     None
//                 }
//             }
//             _ => {
//                 return Err(RedisError::RespDeserialize(
//                     "the type of cluster slot node doesn't be map".to_string(),
//                 ))
//             }
//         };
//         Ok(Self {
//             ip,
//             port,
//             node_id,
//             host,
//         })
//     }
// }

fn get_shard_by_slot(shards: &Vec<ClusterShard>, slot: u16) -> Option<&ClusterShard> {
    for shard in shards {
        if shard.slots.contains(&slot) {
            return Some(shard);
        }
    }
    None
}

#[derive(Debug)]
#[repr(u8)]
enum ClusterRole {
    Master = 0,
    Replica = 1,
}

#[derive(Debug)]
pub(crate) struct ClusterShard {
    // from: u16,
    // to: u16,
    slots: Vec<u16>,
    master: ClusterNode,
    replicas: SmallVec<[ClusterNode; 4]>,
}

impl ClusterShard {
    fn read_from_value(value: Value, original_ip: &str) -> RedisResult<Vec<Self>> {
        let mut result = vec![];
        let mut temp = smallvec::smallvec![];
        let str_value = value.into_string()?;
        let lines = str_value.lines();
        for line in lines {
            let array = line.split_ascii_whitespace();
            Self::internal_read_line(array, &mut result, &mut temp, original_ip)?;
        }
        for item in result.iter_mut() {
            let mut vec = SmallVec::<[usize; 4]>::new();
            for (index, data) in temp.iter().enumerate() {
                if data.master_node_id == item.master.id {
                    vec.push(index)
                }
            }
            vec.sort_by(|a, b| b.cmp(a));
            for num in vec {
                item.replicas.push(temp.remove(num));
            }
        }
        Ok(result)
    }

    fn internal_read_line(
        mut array: SplitAsciiWhitespace,
        shards: &mut Vec<ClusterShard>,
        temp: &mut SmallVec<[ClusterNode; 4]>,
        original_ip: &str,
    ) -> RedisResult<()> {
        let node_id = array.next().ok_or(RedisError::RespDeserialize(
            "couldn't read [node id] from cluster node line".to_string(),
        ))?;
        let host_info = array.next().ok_or(RedisError::RespDeserialize(
            "couldn't read [host info] from cluster node line".to_string(),
        ))?;
        let role = array.next().ok_or(RedisError::RespDeserialize(
            "couldn't read [host info] from cluster node line".to_string(),
        ))?;
        if role.contains("master") {
            // 为主节点
            array.nth(4);
            let mut slots = vec![];
            loop {
                if let Some(str) = array.nth(0) {
                    if str.contains("->-") || str.contains("-<-") {
                        continue;
                    } else if str.contains('-') {
                        let (from, to) = str.split_once('-').ok_or(RedisError::Connect(
                            format!("error on read slot, str = {str}"),
                        ))?;
                        let mut from = from.parse::<u16>().map_err(|err| {
                            RedisError::new_connect(
                                format!("error on parse `from`, str = {str}"),
                                err,
                            )
                        })?;
                        let to = to.parse::<u16>().map_err(|err| {
                            RedisError::new_connect(
                                format!("error on parse `to`, str = {str}"),
                                err,
                            )
                        })?;
                        while from <= to {
                            slots.push(from);
                            from += 1;
                        }
                    } else {
                        slots.push(str.parse::<u16>().map_err(|err| {
                            RedisError::new_connect(format!("error on read slot, str = {str}"), err)
                        })?);
                    }
                } else {
                    break;
                };
            }
            shards.push(ClusterShard {
                slots,
                master: ClusterNode::new_master(node_id.to_string(), host_info, original_ip)?,
                replicas: smallvec::smallvec![],
            });
        } else {
            // 从节点
            let master_node_id = array.next().ok_or(RedisError::RespDeserialize(
                "couldn't read [host info] from cluster node line".to_string(),
            ))?;
            temp.push(ClusterNode::new_replica(
                node_id.to_string(),
                host_info,
                master_node_id.to_string(),
                original_ip,
            )?);
        }
        Ok(())
    }
}

impl PartialEq for ClusterShard {
    fn eq(&self, other: &Self) -> bool {
        self.master.host == other.master.host && self.master.port == other.master.port
    }
}

#[derive(Debug)]
struct ClusterNode {
    id: String,
    host: String,
    port: u32,
    #[allow(dead_code)]
    host_name: String,
    #[allow(dead_code)]
    role: ClusterRole,
    master_node_id: String,
}

impl ClusterNode {
    fn new_master(id: String, host_info: &str, original_ip: &str) -> RedisResult<Self> {
        let (host, port, host_name) = Self::analysis_hostinfo(host_info, original_ip)?;
        Ok(Self {
            id,
            host,
            port,
            host_name,
            role: ClusterRole::Master,
            master_node_id: String::new(),
        })
    }

    fn new_replica(
        id: String,
        host_info: &str,
        master_node_id: String,
        original_ip: &str,
    ) -> RedisResult<Self> {
        let (host, port, host_name) = Self::analysis_hostinfo(host_info, original_ip)?;
        Ok(Self {
            id,
            host,
            port,
            host_name,
            role: ClusterRole::Replica,
            master_node_id,
        })
    }

    fn analysis_hostinfo(host_info: &str, original_ip: &str) -> RedisResult<(String, u32, String)> {
        let (ip, nparts1) = host_info
            .split_once(':')
            .ok_or(RedisError::RespDeserialize(format!(
                "couldn't use `:` to split host info {host_info}"
            )))?;
        let ip = if ip == "127.0.0.1" {
            original_ip.to_string()
        } else {
            ip.to_string()
        };
        let (port, nparts2) =
            nparts1
                .split_once('@')
                .ok_or(RedisError::RespDeserialize(format!(
                    "couldn't use `@` to split host info {host_info}"
                )))?;
        let host = if let Some((_, host)) = nparts2.split_once(",") {
            host.to_string()
        } else {
            ip.clone()
        };
        // let (_, host) = nparts2
        //     .split_once(',')
        //     .ok_or(RedisError::RespDeserialize(format!(
        //         "couldn't use `,` to split host info {host_info}"
        //     )))?;
        Ok((
            ip,
            port.parse::<u32>().map_err(|err| {
                RedisError::new_resp_deserialize(format!("error on parse port:{port} to u32"), err)
            })?,
            host,
        ))
    }
}

pub(crate) fn hash_slot(key: &str) -> u16 {
    let mut key = key;
    if let Some(s) = key.find('{') {
        if let Some(e) = key[s + 1..].find('}') {
            if e != 0 {
                key = &key[s + 1..s + 1 + e];
            }
        }
    }
    crc16::State::<crc16::XMODEM>::calculate(key.as_bytes()) % 16384
}

/// format1: ASK 15495 192.168.128.131:6401
///
/// format2: MOVED 15495 192.168.128.131:6401
fn read_moved_or_ask(str: &str) -> Option<(String, u16, String, u32)> {
    if !str.starts_with("MOVED ") || !str.starts_with("ASK ") {
        return None;
    }
    let mut splits = str.split(' ');
    let action = splits.nth(0)?;
    let slot = splits.nth(0)?;
    let slot = slot.parse::<u16>().ok()?;
    let mut host_info = splits.nth(0)?.split(':');
    let host = host_info.nth(0)?;
    let port = host_info.nth(0)?;
    let port = port.parse::<u32>().ok()?;
    Some((action.to_string(), slot, host.to_string(), port))
}
